package com.alibaba.otter.node.etl.load.loader.mq.rabbit.assembly;

import com.alibaba.otter.node.etl.load.loader.mq.rabbit.MqConstants;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.Header;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.Message;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.provider.*;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.serializer.Serializer;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.rabbitmq.client.AMQP;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Slf4j
public class RabbitMqMessageAssembler implements MessageAssembler {

    @Setter
    private Serializer serializer;

    @Setter
    private String exchange;
    @Setter
    private String routingKey;

    @Setter
    private List<PropertyProvider> propertyProviders = Lists.newArrayList(
        new DefaultDeliveryModeProvider(),
        //new DefaultCorrelationIdProvider(),
        new DefaultTypeProvider(),
        new DefaultClusterProvider(),
        new DefaultTimestampProvider(),
        new DefaultContentEncodingProvider()
    );

    @Setter
    private List<HeaderProvider> headerProviders = Lists.newArrayList(
        new RequestHeaderProvider()
    );

    public void addHeaderProviders(List<HeaderProvider> headerProviders) {
        this.headerProviders.addAll(headerProviders);
    }

    public void addPropertyProviders(List<PropertyProvider> propertyProviders) {
        this.propertyProviders.addAll(propertyProviders);
    }

    @Override
    public Message assemblyMessage(Object messageContent) {

        if (messageContent instanceof Message) {
            return (Message) messageContent;
        }

        Message message = new Message();
        if (serializer.getCharset() == null) {
            List<ContentEncodingProvider> encodingProviders = propertyProviders.stream().
                filter(e -> e instanceof ContentEncodingProvider).map(e -> (ContentEncodingProvider) e).
                collect(Collectors.toList());
            if (CollectionUtils.isNotEmpty(encodingProviders)) {
                // 以最后一个contentEncodingProvider为准。
                ContentEncodingProvider provider = encodingProviders.get(encodingProviders.size() - 1);
                String charsetName = provider.provide(null).toString();
                serializer.setCharset(Charset.forName(charsetName));
            } else {
                serializer.setCharset(StandardCharsets.UTF_8);
            }
        }

        message.setContent(serializer.serialize(messageContent));

        AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
        propertyProviders.forEach(provide -> {
            try {
                Method method = ReflectionUtils.findMethod(
                    AMQP.BasicProperties.Builder.class,
                    provide.propertyName(),
                    provide.propertyType()
                );
                ReflectionUtils.invokeMethod(method, builder, provide.provide(messageContent));
            } catch (Exception e) {
                log.warn("Failed to set property.", e);
            }
        });

        Map<String, Object> headers = Maps.newHashMap(ImmutableMap.of(
            "exchange", this.exchange,
            "routingKey", this.routingKey
        ));

        builder.headers(headers);
        AMQP.BasicProperties properties = builder.build();
        headerProviders.forEach(provider -> {
            try {
                List<Header> providedHeaders = provider.provide(properties, messageContent);
                providedHeaders.forEach(header -> headers.put(header.getName(), header.getValue()));
            } catch (Exception e) {
                log.warn("Failed to set header.", e);
            }
        });
        Map<String, Object> propertyHeaders = Maps.newHashMap(headers);
        MqConstants.IGNORE_HEADERS.forEach(ignoreHeader -> propertyHeaders.remove(ignoreHeader));
        builder.headers(propertyHeaders);
        AMQP.BasicProperties properties2 = builder.build();
        message.setProperty(properties2);
        message.setHeaders(headers);
        return message;
    }
}
